package com.yk.queue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @Date: 2018/11/12 17:38
 * @Version: 1.0
 * @Author: pengqingfeng
 * @Description:
 */
@Component
public class KafkaSender {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());


    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;


    // 发送消息方法
    public void send(String channal, String key, String body) {
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(channal, key, body);
        future.addCallback(new myCallback());
//        logger.info("发送消息完成,内容为: key:{}----body:{}", key, body);
    }

    // 发送消息方法
    public void send(String channal, String body) {
        kafkaTemplate.send(channal, body);
        logger.info("发送消息完成,内容为: body:{}", body);
    }
}

class myCallback implements ListenableFutureCallback<SendResult<String, Object>> {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Override
    public void onFailure(Throwable throwable) {
        logger.error("发送失败：" + throwable.getMessage());
    }

    @Override
    public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
        logger.info("发送成功：" + stringObjectSendResult.toString());
    }
}